package demo.db.elasticSearch.transport;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.lucene.queryparser.classic.QueryParser;
import org.apache.lucene.search.TotalHits;
import org.elasticsearch.action.ActionFuture;
import org.elasticsearch.action.admin.cluster.health.ClusterHealthRequest;
import org.elasticsearch.action.admin.cluster.health.ClusterHealthResponse;
import org.elasticsearch.action.admin.indices.create.CreateIndexResponse;
import org.elasticsearch.action.bulk.BulkRequestBuilder;
import org.elasticsearch.action.get.GetRequestBuilder;
import org.elasticsearch.action.get.GetResponse;
import org.elasticsearch.action.index.IndexRequestBuilder;
import org.elasticsearch.action.search.SearchRequestBuilder;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.update.UpdateRequestBuilder;
import org.elasticsearch.client.Client;
import org.elasticsearch.client.IndicesAdminClient;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.transport.TransportAddress;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.common.xcontent.XContentBuilder;
import org.elasticsearch.common.xcontent.XContentFactory;
import org.elasticsearch.index.query.BoolQueryBuilder;
import org.elasticsearch.index.query.MatchPhraseQueryBuilder;
import org.elasticsearch.index.query.QueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.index.reindex.UpdateByQueryRequestBuilder;
import org.elasticsearch.script.Script;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.SearchHits;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.elasticsearch.transport.client.PreBuiltTransportClient;
import org.junit.Test;
import org.springframework.beans.factory.annotation.Value;

import java.io.BufferedReader;
import java.io.FileReader;
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

import static org.elasticsearch.index.query.QueryBuilders.matchPhraseQuery;

/**
 * ES提供了两个JAVA REST client 版本
 * <li>Java Low Level REST Client:
 * 低级别的REST客户端，通过http与集群交互，用户需自己编组请求JSON串，及解析响应JSON串。兼容所有ES版本。
 * <li>Java High Level REST Client:
 * 高级别的REST客户端，基于低级别的REST客户端，增加了编组请求JSON串、解析响应JSON串等相关api。使用的版本需要保持和ES服务端的版本一致，否则会有版本问题。
 *
 * @author hanjy
 */
public class EsDemo {

    private TransportClient transportClient;

    @Value("${es.connection.host}")
    private static String hosts = "192.168.60.97:9300,192.168.60.98:9300,192.168.60.99:9300";

    @Value("${es.cluster.name}")
    private static String clusterName = "xdcs-cluster";

    @Value("${es.conflict.retry.times}")
    private Integer retry;

    public static final int FROM = 0;
    public static final int SIZE = 5000;

    /**
     * 健康检查
     */
    @Test
    public void health() throws ExecutionException, InterruptedException, TimeoutException, UnknownHostException {
        try (TransportClient client = init(clusterName, hosts)) {
            ClusterHealthRequest request = new ClusterHealthRequest();
            ActionFuture<ClusterHealthResponse> future = client.admin().cluster().health(request);
            ClusterHealthResponse response = future.get(60, TimeUnit.SECONDS);
            System.err.println(JSON.toJSONString(response, true));
            System.err.println(response.getClusterName() + " : " + response.getStatus().name());
        }
    }

    /**
     * 创建空索引 默认setting 无mapping
     *
     * @param client
     * @param index
     * @return
     */
    public boolean createSimpleIndex(Client client, String index) {
        IndicesAdminClient indicesAdminClient = client.admin().indices();
        CreateIndexResponse response = indicesAdminClient.prepareCreate(index).get();
        return response.isAcknowledged();
    }

    /**
     * 创建索引 指定setting,创建mapper
     *
     * @param client
     * @param index
     * @return
     */
    public boolean createIndex(Client client, String index) {
        // settings
        Settings settings = Settings.builder().put("index.number_of_shards", 1).put("index.number_of_replicas", 0)
                // .put("cluster.name", "poc")
                // .put("node.name", "node1")
                // .put("client.transport.ignore_cluster_name", true)
                // .put("node.client", true)
                // .put("client.transport.sniff", true)
                .build();
        // mapping
        XContentBuilder mappingBuilder;
        try {
            mappingBuilder = XContentFactory.jsonBuilder().startObject().startObject(index).startObject("properties")
                    .startObject("ID").field("type", "string").field("store", "yes").endObject().startObject("IP")
                    .field("type", "string").field("store", "yes").endObject().startObject("VALUE_DAY")
                    .field("type", "double").field("store", "yes").endObject().startObject("GROUPNAME")
                    .field("type", "string").field("store", "yes").endObject().startObject("TYPE_MACHINE")
                    .field("type", "string").field("store", "yes").endObject().startObject("DATE")
                    .field("type", "string").field("store", "yes").endObject().startObject("TYPE")
                    .field("type", "string").field("store", "yes").endObject().endObject().endObject().endObject();
        } catch (Exception e) {
            System.out.println("--------- createIndex 创建 mapping 失败：" + e);
            return false;
        }
        IndicesAdminClient indicesAdminClient = client.admin().indices();
        CreateIndexResponse response = indicesAdminClient.prepareCreate(index).setSettings(settings)
                .addMapping(index, mappingBuilder).get();
        return response.isAcknowledged();
    }

    /**
     * 批量导出
     */
    public SearchHits bulkOut(Client client, String index, String type) {

        SearchResponse response = client.prepareSearch(index).setTypes(type).setQuery(QueryBuilders.matchAllQuery())
                .execute().actionGet();
        SearchHits resultHits = response.getHits();

        return resultHits;
    }

    /**
     * 批量导入
     *
     * @throws Exception
     */
    public void bulkIn(Client client, String index, String type, String filePath, String[] columnNameList, String token)
            throws Exception {

        List<String> dataList = dataCreater(filePath, columnNameList, token);

        BulkRequestBuilder bulkRequest = client.prepareBulk();

        for (String data : dataList) {
            bulkRequest.add(client.prepareIndex(index, type).setSource(data));
        }
        bulkRequest.execute().actionGet();
    }

    /**
     * 查询全部内容，实现分页查询
     */
    public List<String> searchAll(Client client, String index, String table) {

        SearchResponse response = client.prepareSearch().setIndices(index).setTypes(table).get();
        long n = response.getHits().getTotalHits().value;
        int size = 100;
        List<String> retList = new ArrayList<String>();

        for (int i = 0; i < n; ) {
            response = client.prepareSearch().setIndices(index).setTypes(table).setScroll(TimeValue.timeValueMinutes(2))
                    .setFrom(i).setSize(size).get();
            i += size;
            SearchHits searchHits = response.getHits();
            for (SearchHit hit : searchHits) {
                retList.add(hit.getSourceAsString());
            }
        }
        return retList;
    }

    /**
     * 使用过滤器查询，实现分页查询
     */
    public List<String> queryByFilter(Client client, String index, String table) {

        // 查询groupname为"压力测试"的数据
        QueryBuilder queryBuilder = QueryBuilders.matchQuery("GROUPNAME", "压力测试");

        SearchResponse response = client.prepareSearch().setIndices(index).setTypes(table).setQuery(queryBuilder).get();
        TotalHits n = response.getHits().getTotalHits();
        System.out.println(n);
        int size = 100;
        List<String> retList = new ArrayList<String>();

        response = client.prepareSearch().setIndices(index).setTypes(table).setScroll(TimeValue.timeValueMinutes(5))
                .setSize((int) n.value).setQuery(queryBuilder).get();
        SearchHits searchHits = response.getHits();
        for (SearchHit hit : searchHits) {
            retList.add(hit.getSourceAsString());
        }

        return retList;
    }

    /**
     * 数据写入索引
     *
     * @param client
     * @param index
     * @return
     * @throws Exception
     */
    public void write(Client client, String index, String tablename, String filePath, String[] columnNameList,
                      String token) throws Exception {

        // 创建索引库 需要注意的是.setRefresh(true)这里一定要设置,否则第一次建立索引查找不到数据
        IndexRequestBuilder requestBuilder = client.prepareIndex(index, tablename);
        List<String> dataList = dataCreater(filePath, columnNameList, token);
        for (String data : dataList) {
            requestBuilder.setSource(data).execute().actionGet();
        }
    }

    /**
     * @param clusterName 默认集群名为elasticsearch，如果集群名称和指定的不一致则在使用节点资源时会报错。
     * @param hosts
     * @return
     * @throws UnknownHostException
     */
    public static TransportClient init(String clusterName, String hosts) throws UnknownHostException {
        System.setProperty("es.set.netty.runtime.available.processors", "false");
        Settings settings = Settings.builder().put("cluster.name", clusterName).build();
        TransportClient transportClient = new PreBuiltTransportClient(settings);
        String[] hostArray = hosts.split(",");

        for (String host : hostArray) {
            String[] param = host.split(":");
            transportClient.addTransportAddress(
                    new TransportAddress(InetAddress.getByName(param[0]), Integer.parseInt(param[1])));
        }
        return transportClient;
    }

    /**
     * insert or update by docId
     *
     * @param index
     * @param type
     * @param id
     * @param param
     */
    public void upsert(String index, String type, String id, Map<String, Object> param) {
        UpdateRequestBuilder urb = transportClient.prepareUpdate(index, type, id);
        urb.setRetryOnConflict(retry);
        urb.setDoc(param);
        urb.setDocAsUpsert(Boolean.TRUE);
        urb.execute().actionGet();
    }

    /**
     * update by multi params
     *
     * @param index
     * @param filter
     * @param now
     */
    public void updateByParam(String index, Map<String, Object> filter, Map<String, Object> now) {
        UpdateByQueryRequestBuilder builder = null;
//        UpdateByQueryAction.INSTANCE.newRequestBuilder(transportClient);
        builder = builder.source(index).abortOnVersionConflict(Boolean.FALSE);
        Iterator<String> keyIt = filter.keySet().iterator();
        while (keyIt.hasNext()) {
            String key = keyIt.next();
            Object value = filter.get(key);
            builder.filter(matchPhraseQuery(key, value));
        }
        builder.script(buildScript(now)).get();
    }

    /**
     * delete by docId
     *
     * @param index
     * @param type
     * @param docId
     * @throws IllegalAccessException
     */
    public void deleteById(String index, String type, String docId) {
        transportClient.prepareDelete(index, type, docId).get();
    }

    /**
     * query data according multi params
     *
     * @param index
     * @param type
     * @param params
     * @return
     * @throws IllegalAccessException
     */
    public List<Map<String, Object>> queryAndParam(String index, String type, Map<String, Object> params) {
        SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
        QueryBuilder queryBuilder = QueryBuilders.boolQuery();

        Iterator<String> keyIt = params.keySet().iterator();
        while (keyIt.hasNext()) {
            String key = keyIt.next();
            Object value = params.get(key);
            if (value instanceof String) {
                value = QueryParser.escape(value.toString());
            }
            MatchPhraseQueryBuilder filter = matchPhraseQuery(key, value);

            ((BoolQueryBuilder) queryBuilder).must(filter);
        }
        sourceBuilder.query(queryBuilder);
        SearchRequestBuilder searchRequestBuilder = transportClient.prepareSearch(index).setTypes(type);
        return response(searchRequestBuilder.setQuery(queryBuilder).execute().actionGet());
    }

    /**
     * 单 field 多值 or 查询方法
     *
     * @param index
     * @param type
     * @param fields
     * @param matchString
     * @return
     */
    public List<Map<String, Object>> queryOrParam(String index, String type, List<String> fields, String matchString) {
        SearchRequestBuilder searchRequestBuilder = transportClient.prepareSearch(index).setTypes(type);
        BoolQueryBuilder boolQueryBuilder = QueryBuilders.boolQuery();
        for (String field : fields) {
            boolQueryBuilder.should(QueryBuilders.queryStringQuery(QueryParser.escape(matchString)).field(field));
        }
        searchRequestBuilder.setQuery(boolQueryBuilder);
        searchRequestBuilder.setFrom(FROM).setSize(SIZE);
        return response(searchRequestBuilder.execute().actionGet());
    }

    /**
     * 对Map中的key、value执行Or查询
     *
     * @param index
     * @param type
     * @param params
     * @return
     */
    public List<Map<String, Object>> queryOrMap(String index, String type, Map<String, String> params) {
        SearchRequestBuilder searchRequestBuilder = transportClient.prepareSearch(index).setTypes(type);
        BoolQueryBuilder boolQueryBuilder = QueryBuilders.boolQuery();
        for (Entry<String, String> e : params.entrySet()) {
            boolQueryBuilder.should(QueryBuilders.queryStringQuery(QueryParser.escape(e.getValue())).field(e.getKey()));
        }
        searchRequestBuilder.setQuery(boolQueryBuilder);
        searchRequestBuilder.setFrom(FROM).setSize(SIZE);
        return response(searchRequestBuilder.execute().actionGet());
    }

    /**
     * query by docId
     *
     * @param index
     * @param type
     * @param id
     * @return
     * @throws IllegalAccessException
     */
    public Map<String, Object> queryById(String index, String type, String id) {
        GetRequestBuilder req = transportClient.prepareGet(index, type, id);
        GetResponse resp = req.get();
        if (resp.isExists() && (!resp.isSourceEmpty())) {
            return resp.getSourceAsMap();
        }
        return null;
    }

    private List<Map<String, Object>> response(SearchResponse response) {
        SearchHits hits = response.getHits();
        TotalHits totalHits = hits.getTotalHits();
        long size = totalHits.value;
        List<Map<String, Object>> list = (size <= 0) ? new ArrayList<>(0) : new ArrayList<>((int) size);
        for (SearchHit searchHit : response.getHits()) {
            list.add(searchHit.getSourceAsMap());
        }
        return list;
    }

    public static Script buildScript(Map<String, Object> now) {
        StringBuilder builder = new StringBuilder();
        Iterator<String> keyIt = now.keySet().iterator();
        while (keyIt.hasNext()) {
            String key = keyIt.next();
            Object value = now.get(key);
            builder.append("ctx._source.");
            builder.append(key);
            builder.append("=\"");
            builder.append(value);
            builder.append("\";");
        }
        String script = builder.toString();
        script = script.substring(0, script.length() - 1);
        return new Script(script);
    }

    public static List<String> dataCreater(String path, String[] columnNameList, String token) throws Exception {

        FileReader fr = new FileReader(path);
        BufferedReader br = new BufferedReader(fr);
        String line = null;
        List<String> dataList = new ArrayList<String>();

        while ((line = br.readLine()) != null) {
            String[] data = line.split(token);
            Map<String, Object> map = new HashMap<String, Object>();
            for (int i = 0; i < columnNameList.length; i++) {
                map.put(columnNameList[i], data[i]);
            }
            JSONObject jsonObject = new JSONObject(map);
            String str = jsonObject.toString();
            dataList.add(str);
        }

        br.close();
        fr.close();

        return dataList;
    }
}